package workmanager

import (
	"sync"
	"fmt"
	"io/ioutil"
	"zipfile"
	"path/filepath"
	"time"
	"process"
	"kafka"
	"log"
)

type WorkPara struct {
	wd string
	bl string
	tp string
	fq int
}

type WorkManager struct {
	num int
	ch chan bool
	waitGroup *sync.WaitGroup
	wp WorkPara
}

func NewWorkManager(num int,wd,bl,tp string,fq int)*WorkManager{
	wm := new(WorkManager)
	wm.num = num
	wm.ch = make(chan bool)
	wm.waitGroup = new(sync.WaitGroup)
	wm.wp.wd = wd
	wm.wp.bl = bl
	wm.wp.tp = tp
	wm.wp.fq = fq
	return wm
}

func (wm WorkManager)Start(){
	workch := make(chan string,20)
	stopch := make(chan bool,wm.num+1)

	go wm.Producer(workch,stopch)
	for i := 0; i< wm.num; i++{
		go wm.Consumer(workch,stopch)
	}

	<- wm.ch
	for i:=0;i<wm.num+1;i++{
		stopch <- true
	}
	wm.wait()
	fmt.Println("stop gracefully.")
}

func (wm WorkManager)Stop(){
	wm.ch <- true
	wm.wait()
}

func (wm WorkManager)Consumer(ch chan string, stop chan bool){
	wm.waitGroup.Add(1)
	k,err := kafka.NewKafka(wm.wp.bl,wm.wp.tp)
	if err !=nil{
		log.Println("kafka connect error, broker: ",wm.wp.bl,", topic: ",wm.wp.tp)
		wm.waitGroup.Done()
		return
	}

Loop:
	for{
		select {
		case fp:= <- ch:
			p := process.NewFileProcess(fp)
			p.FileProcess(k)
			log.Println(fp)
		case <- stop:
			break Loop
		default:

		}
	}
	fmt.Println("stop comsumer")
	wm.waitGroup.Done()
}


func (wm WorkManager)Producer(ch chan string,stop chan bool){
	wm.waitGroup.Add(1)
	timeout := make(chan bool)



Loop:
	for{
		go func(fq int) {
			sleeptime := 1
			if fq != 0{
				sleeptime = fq
			}
			time.Sleep(time.Duration(sleeptime)*60*time.Second)
			timeout <- true
		}(wm.wp.fq)
		files, _ := ioutil.ReadDir(wm.wp.wd)

		for _, fp := range files{

			if fp.IsDir(){
				continue
			}

			abspath := filepath.Join(wm.wp.wd,fp.Name())
			if ret := zipfile.ValidZipFile(abspath);ret == false{
				continue
			}
			select {
			case ch <- abspath:
				continue
			case <-stop:
				break Loop
			}

		}
		select {
		case <- stop:
			close(ch)
			break Loop
		case <- timeout:
			continue
		}

	}
	fmt.Println("stop producer")
	wm.waitGroup.Done()
}

func (wm WorkManager)wait(){
	wm.waitGroup.Wait()
}